package kafka.service;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.domain.LyUserchat;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.JDBCUntil;
import kafka.utils.VerifiableProperties;
import net.sf.json.JSONObject;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.*;

//@Component
//@Order(value=2)
public class KafkaConsumer {

    private final ConsumerConnector consumer;
    public final static String TOPIC = "TEST-TOPIC";

    public KafkaConsumer() {
        Properties props = new Properties();
        //zookeeper 配置
//        props.put("zookeeper.connect", "192.168.1.101:2181");
        props.put("zookeeper.connect", "www.dayu90du.top:2181");

        //group 代表一个消费组
        props.put("group.id", "jd-group");

        //zk连接超时
        props.put("zookeeper.session.timeout.ms", "4000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.offset.reset", "largest");
        //序列化类
        props.put("serializer.class", "kafka.serializer.StringEncoder");

        ConsumerConfig config = new ConsumerConfig(props);

        consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);

    }

    public void consume() {
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(this.TOPIC, new Integer(1));

        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());

        Map<String, List<KafkaStream<String, String>>> consumerMap =
                consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
        KafkaStream<String, String> stream = consumerMap.get(this.TOPIC).get(0);
        ConsumerIterator<String, String> it = stream.iterator();
        while (it.hasNext()) {

            try {
                String objstr = it.next().message();
                System.out.println("消费者：" + objstr);

                JSONObject jsonObject = JSONObject.fromObject(objstr);
                LyUserchat lyUserchat = new LyUserchat();
                lyUserchat.setSenderId(String.valueOf(jsonObject.get("senderId")));
                lyUserchat.setReciverId(String.valueOf(jsonObject.get("reciverId")));
                System.out.println(Timestamp.valueOf(String.valueOf(jsonObject.get("createTime"))));
                lyUserchat.setCreateTime(Timestamp.valueOf(String.valueOf(jsonObject.get("createTime"))));
                lyUserchat.setContent(String.valueOf(jsonObject.get("content")));

                insertChatdata(lyUserchat);

            } catch (Exception e) {
                e.printStackTrace();
            }

        }
    }


    public int insertChatdata(LyUserchat lyUserchat) {
        Connection conn = null;
        try {
            conn = JDBCUntil.getConnection();
        } catch (SQLException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        int i = 0;
        String sql = "INSERT INTO ly_userchat(sender_id,reciver_id,create_time,content) VALUES (?,?,?,?)";
        PreparedStatement pstmt;
        try {
            pstmt = (PreparedStatement) conn.prepareStatement(sql);
            pstmt.setString(1, lyUserchat.getSenderId());
            pstmt.setString(2, lyUserchat.getReciverId());
            pstmt.setTimestamp(3, lyUserchat.getCreateTime());
            pstmt.setString(4, lyUserchat.getContent());
            i = pstmt.executeUpdate();
            pstmt.close();
            conn.close();

        } catch (SQLException e) {
            e.printStackTrace();
        }
        return i;
    }

//    @PostConstruct
//    public void run123(){
//        new KafkaConsumer();
//    }

//    public static void main(String[] args) {
//        new KafkaConsumer().consume();
//    }
}
